package cn.edu.zut.qixiang.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.time.LocalDateTime;
import java.util.Properties;

/**
 * @author 86178
 */
public class WeatherAnalysisJob {

    private static final ObjectMapper mapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 配置Kafka源
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
        properties.setProperty("group.id", "weather-analysis-group");
        properties.setProperty("auto.offset.reset", "earliest");


        DataStream<String> kafkaStream = env.addSource(
                new FlinkKafkaConsumer<>(
                        "maxwell_weather_data",
                        new SimpleStringSchema(),
                        properties
                )
        );
        // 解析JSON数据
        SingleOutputStreamOperator<WeatherEvent> weatherStream = kafkaStream
                .map(new MaxwellJsonParser())
                .name("Parse Maxwell JSON");

        // 将原始数据转发到新的Kafka topic供前端使用
        SingleOutputStreamOperator<String> forwardedData = weatherStream
                .map(new ObjectMapper()::writeValueAsString)
                .name("Forward Raw Data");

        // 转发到新的topic
        forwardedData
                .addSink(new FlinkKafkaProducer<>(
                        "flink_processed_weather_data",
                        new SimpleStringSchema(),
                        properties
                ))
                .name("Forward to Frontend Topic");
        // 实时分析1: 各区域温度统计
        SingleOutputStreamOperator<TemperatureStats> tempStats = weatherStream
                .keyBy(WeatherEvent::getDistrict)
                .timeWindow(Time.minutes(5))
                .aggregate(new TemperatureAggregator(), new TemperatureWindowFunction())
                .name("Temperature Statistics");

        // 实时分析2: 异常天气检测
        SingleOutputStreamOperator<WeatherAlert> alerts = weatherStream
                .filter(event -> event.getTemperature() > 35.0 || event.getWindSpeed() > 10.0)
                .map(event -> new WeatherAlert(
                        event.getDistrict(),
                        event.getTemperature(),
                        event.getWindSpeed(),
                        LocalDateTime.now()
                ))
                .name("Weather Alerts");

        // 输出结果（这里可以输出到Kafka、文件或数据库）
        tempStats.print("温度统计");
        alerts.print("天气预警");

        // 执行任务
        env.execute("Real-time Weather Analysis");
    }

    // JSON解析器
    public static class MaxwellJsonParser implements MapFunction<String, WeatherEvent> {
        @Override
        public WeatherEvent map(String value) throws Exception {
            JsonNode jsonNode = mapper.readTree(value);

            // 只处理insert操作
            if (!"insert".equals(jsonNode.get("type").asText())) {
                return null;
            }

            JsonNode data = jsonNode.get("data");
            WeatherEvent event = new WeatherEvent();
            event.setDistrict(data.get("district").asText());
            event.setTimestamp(data.get("timestamp").asText());
            event.setTemperature(data.get("temperature").asDouble());
            event.setHumidity(data.get("humidity").asInt());
            event.setWindSpeed(data.get("wind_speed").asDouble());
            event.setPressure(data.get("pressure").asDouble());
            event.setPrecipitation(data.get("precipitation").asDouble());

            return event;
        }
    }

    // 温度聚合器
    public static class TemperatureAggregator implements AggregateFunction<
            WeatherEvent, TemperatureAccumulator, TemperatureAccumulator> {

        @Override
        public TemperatureAccumulator createAccumulator() {
            return new TemperatureAccumulator();
        }

        @Override
        public TemperatureAccumulator add(WeatherEvent event, TemperatureAccumulator accumulator) {
            accumulator.addTemperature(event.getTemperature());
            accumulator.setDistrict(event.getDistrict());
            return accumulator;
        }

        @Override
        public TemperatureAccumulator getResult(TemperatureAccumulator accumulator) {
            return accumulator;
        }

        @Override
        public TemperatureAccumulator merge(TemperatureAccumulator a, TemperatureAccumulator b) {
            a.merge(b);
            return a;
        }
    }

    // 温度窗口函数
    public static class TemperatureWindowFunction extends ProcessWindowFunction<
            TemperatureAccumulator, TemperatureStats, String, TimeWindow> {

        @Override
        public void process(String district, Context context,
                            Iterable<TemperatureAccumulator> elements,
                            Collector<TemperatureStats> out) {
            TemperatureAccumulator accumulator = elements.iterator().next();
            out.collect(new TemperatureStats(
                    district,
                    accumulator.getAvgTemperature(),
                    accumulator.getMaxTemperature(),
                    accumulator.getMinTemperature(),
                    accumulator.getCount(),
                    LocalDateTime.now()
            ));
        }
    }
}